package com.cryann.mqttserver.service.Impl;

import com.cryann.mqttserver.enums.ConfirmStatus;
import com.cryann.mqttserver.modle.MqttMessageStore;
import com.cryann.mqttserver.modle.MqttSendMessage;
import com.cryann.mqttserver.service.MsgStoreService;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Queue;

/**
 * @author cRyann
 * @date 2018/12/21 0021 16:40
 * @description:
 */
@Service("msgStoreService")
@Slf4j
public class MsgStoreServiceImpl implements MsgStoreService {

    @Autowired
    MqttMessageStore mqttMessageStore;

    @Override
    public MqttSendMessage storeMsg(Channel channel, MqttPublishMessage message) {
        try {
            MqttSendMessage sendMqttMessage = conversionMsg(channel, message);
            mqttMessageStore.addMsg(channel, sendMqttMessage);
            return sendMqttMessage;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 返回队列头部的元素 如果队列为空，则返回null
     **/
    @Override
    public MqttSendMessage peekMsg(Channel channe) {
        try {
            String key = channe.id().asShortText();
            Queue<MqttSendMessage> queue = mqttMessageStore.getMq(key);
            log.info("[{}] Size[{}]", key);
            return queue.peek();
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 移除并返问队列头部的元素    如果队列为空，则返回null
     **/
    @Override
    public MqttSendMessage pollMsg(Channel channe) {
        try {
            String key = channe.id().asShortText();
            Queue<MqttSendMessage> queue = mqttMessageStore.getMq(key);
            log.info("[{}]", key);
            if ( queue != null ) {
                return queue.poll();
            } else {
                return null;
            }
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    @Override
    public MqttSendMessage conversionMsg(Channel channel, MqttPublishMessage message) {
        MqttSendMessage sendMqttMessage = new MqttSendMessage();
//            sendMqttMessage.setCrcLen(0);
//            sendMqttMessage.setCrc(new byte[]());
        sendMqttMessage.setMessageId(message.variableHeader().packetId());
        sendMqttMessage.setChannel(channel);
        sendMqttMessage.setConfirmStatus(ConfirmStatus.PUB);
        sendMqttMessage.setTime(System.currentTimeMillis());

        ByteBuf byteBuf = message.payload();
        byte[] byteArray = new byte[byteBuf.capacity()];
        byteBuf.readBytes(byteArray);

        sendMqttMessage.setByteBuf(byteArray);
        sendMqttMessage.setRetain(message.fixedHeader().isRetain());
        sendMqttMessage.setQos(message.fixedHeader().qosLevel());
        sendMqttMessage.setTopic(message.variableHeader().topicName());
        sendMqttMessage.setDup(false);
        sendMqttMessage.setMessageType(message.fixedHeader().messageType());
        sendMqttMessage.setRemainingLength(0);
        return sendMqttMessage;
    }


}
